package com.qys.livesMall.kafka.handle;


import com.qys.livesMall.kafka.entity.ExpressInquiryVO;
import com.qys.livesMall.kafka.entity.log.SystemLog;
import com.qys.livesMall.kafka.entity.shop.ShopBrowseRecord;
import com.qys.livesMall.kafka.service.RecordHandleServer;
import com.qys.livesMall.kafka.service.SystemLogService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.errors.SerializationException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.PropertySource;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;

/**
 * kafka监听消息
 */
@Slf4j
@Component
@PropertySource(value = "classpath:config/kafka-topic-${spring.profiles.active}.properties")// 用来指定配置文件的位置
public class KafkaHandler {

    @Autowired
    private RecordHandleServer recordHandleServer;

    @Resource
    private SystemLogService systemLogService;


    private final KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    public KafkaHandler(KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry) {
        this.kafkaListenerEndpointRegistry = kafkaListenerEndpointRegistry;
    }

    /**
     * 监听kafka消息
     *
     * @param consumerRecord kafka的消息，用consumerRecord可以接收到更详细的信息，也可以用String message只接收消息
     * @param ack            kafka的消息确认
     *                       使用autoStartup = "false"必须指定id
     */
    @KafkaListener(topics = {"#{'${shop.browse.records.topic}'}"}, errorHandler = "myKafkaListenerErrorHandler")
    public void shopBrowseRecordListen(ConsumerRecord<Object, ShopBrowseRecord> consumerRecord, Acknowledgment ack) {
        try {
            recordHandleServer.handleShopBrowseRecord(consumerRecord.value());
            //手动确认
            ack.acknowledge();
        } catch (Exception e) {
            if (e instanceof SerializationException) {// 拆解出：xx-topic-0 at offset 55920String s = e.getMessage().split("Error deserializing key/value for partition ")[1].split(". If needed, please seek past the record to continue consumption.")[0];// 拆解 at offset ，之前是 「topic名称-分区编号」；后面是「错误偏移值」String [] ss = s.split("at offset");// 消费的 topic 值String topics = "xx-topic";// 解析分区编号和下一个 offsetint nextOffset = Integer.valueOf(ss[1].trim()) + 1;// topic 分区编号解析int partition = Integer.valueOf(ss[0].trim().replace(topics+"-", ""));// 重新定位到下一个消息，否则这里会一直死循环，即传说中的 Kafka 毒丸TopicPartition topicPartition = new TopicPartition(topics, partition);consumer.seek(topicPartition, nextOffset);
            } else {// TODO 其他异常处理，不影响程序运行
            }
            System.out.println("消费失败：" + e);
        }
    }

    /**
     * 后台操作日志监听
     */
    @KafkaListener(topics = {"#{'${system.log.records.topic}'}"}, errorHandler = "myKafkaListenerErrorHandler")
    public void systemLogListener(ConsumerRecord<Object, SystemLog> consumerRecord, Acknowledgment ack) {
        try {
            systemLogService.save(consumerRecord.value());
            //手动确认
            ack.acknowledge();
        } catch (Exception e) {
            log.error("【系统日志消息消费出错】------》", e);
        }
    }

    /**
     * 快递消息监听
     */
    @KafkaListener(topics = {"#{'${system.express.message.topic}'}"}, errorHandler = "myKafkaListenerErrorHandler")
    public void expressListener(ConsumerRecord<Object, List<ExpressInquiryVO>> consumerRecord, Acknowledgment ack) {
        try {
            log.info("has express message:{}",consumerRecord.value());
            ack.acknowledge();
        } catch (Exception e) {
            log.error("【快递消息消费出错】------》", e);
        }
    }


    /**
     * 下面的方法可以手动操控kafka的队列监听情况
     * 先发送一条消息，因为autoStartup = "false"，所以并不会看到有消息进入监听器。
     * 接着启动监听器，/start/webGroup。可以看到有一条消息进来了。
     * pause是暂停监听，resume是继续监听
     *
     * @param listenerId consumer的group-id
     */
    public void stop(String listenerId) {
        Objects.requireNonNull(kafkaListenerEndpointRegistry.getListenerContainer(listenerId)).pause();
    }

    public void resume(String listenerId) {
        Objects.requireNonNull(kafkaListenerEndpointRegistry.getListenerContainer(listenerId)).resume();
    }

    public void start(String listenerId) {
        Objects.requireNonNull(kafkaListenerEndpointRegistry.getListenerContainer(listenerId)).start();
    }
}
